package com.tj.bigdata.syncblock.strategy;

import com.tj.bigdata.syncblock.config.Web3jConfig;
import com.tj.bigdata.syncblock.constant.Web3jConstants;
import com.tj.bigdata.syncblock.entity.TransactionsStreamSource;
import com.tj.bigdata.syncblock.mapper.DappMapper;
import com.tj.bigdata.syncblock.mapper.ErrorBlockMapper;
import com.tj.bigdata.syncblock.producer.BlockChainInfoKafkaSender;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import org.web3j.protocol.Web3j;
import org.web3j.protocol.core.DefaultBlockParameter;
import org.web3j.protocol.core.methods.response.EthBlock;

import javax.annotation.PostConstruct;
import java.io.IOException;
import java.math.BigInteger;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 流处理数据源
 * ES数据库
 * tx时间戳
 *
 * @author guoch
 */
@Slf4j
@Service
public class StreamSourceAndEsImpl implements SyncBlockStrategy {

    private Web3jConfig web3jConfig;

    private ErrorBlockMapper errorBlockMapper;

    private StringRedisTemplate stringRedisTemplate;

    public static Web3j web3;

    private ThreadPoolExecutor executorService;

    private BlockChainInfoKafkaSender blockChainInfoKafkaSender;

    private DappMapper dappMapper;

    @Autowired
    public void setStringRedisTemplate(StringRedisTemplate stringRedisTemplate) {
        this.stringRedisTemplate = stringRedisTemplate;
    }

    @Autowired
    public void setDappMapper(DappMapper dappMapper) {
        this.dappMapper = dappMapper;
    }

    @Autowired
    public void setBlockChainInfoKafkaSender(BlockChainInfoKafkaSender blockChainInfoKafkaSender) {
        this.blockChainInfoKafkaSender = blockChainInfoKafkaSender;
    }

    @Autowired
    public void setExecutorService(ThreadPoolExecutor executorService) {
        this.executorService = executorService;
    }

    @Autowired
    public void setWeb3jConfig(Web3jConfig web3jConfig) {
        this.web3jConfig = web3jConfig;
    }

    @Autowired
    public void setErrorBlockMapper(ErrorBlockMapper errorBlockMapper) {
        this.errorBlockMapper = errorBlockMapper;
    }

    @PostConstruct
    public void init() {
        web3 = web3jConfig.getWeb3j();
        log.info("以太坊初始化连接成功");
    }

    @Override
    public void batchBlockchain() {

    }

    @Override
    public void batchBlockchain(BigInteger begin, BigInteger end) {

        log.info("开始节点是 {}====》", begin);
        web3.replayPastBlocksFlowable(DefaultBlockParameter.valueOf(begin), DefaultBlockParameter.valueOf(end), true)
                .retry(3).timeout(10, TimeUnit.SECONDS)
                .doOnError(e -> log.error("on error:{}", e.getMessage()))
                .subscribe(this::executeBlock);
    }

    public void execute(Runnable r) {
        executorService.execute(r);
    }

    public void executeBlock(EthBlock block) {
        // 获取到所需的块信息
        EthBlock.Block ethBlock = block.getBlock();
        BigInteger number = ethBlock.getNumber();
        stringRedisTemplate.opsForValue().set(Web3jConstants.BLOCK_FLAG, String.valueOf(number));
        // transaction信息获取
        executeTransaction(ethBlock.getTransactions(), ethBlock.getTimestamp());
    }

    public void executeTransaction(List<EthBlock.TransactionResult> transactions, BigInteger timeStamp) {
        if (transactions.size() == 0) {
            return;
        }
        CountDownLatch countDownLatch = new CountDownLatch(transactions.size());
        for (EthBlock.TransactionResult<EthBlock.TransactionObject> transactionResult : transactions) {
            execute(() -> {
                countDownLatch.countDown();
                EthBlock.TransactionObject transaction = transactionResult.get();
                TransactionsStreamSource streamSource = new TransactionsStreamSource();
                //发送Kafka
                streamSource.setBlockNumber(transaction.getBlockNumber()).setFrom(transaction.getFrom()).setTimestamp(timeStamp).setTo(transaction.getTo()).setValue(transaction.getValue());
                Integer exist = dappMapper.isExist(streamSource.getTo());
                if (exist != 0) {
                    Boolean aBoolean = Optional.of(isContractAddress(streamSource.getFrom())).orElse(false);
                    if (aBoolean) {
                        blockChainInfoKafkaSender.sendTransactionStreamSource(streamSource);
                    }
                    log.info("时间戳是: {} 合约地址是:  {}", streamSource.getTimestamp(), streamSource.getTo());
                }
            });
        }
        try {
            countDownLatch.await();
        } catch (InterruptedException ignored) {
        }
    }

    /**
     * 判断地址是否为合约地址 0x为普通地址
     *
     * @param address 地址
     * @return true 普通地址
     */
    private Boolean isContractAddress(String address) {
        boolean flag;
        try {
            String code = web3.ethGetCode(address, DefaultBlockParameter.valueOf("latest")).send().getCode();
            flag = Web3jConstants.ADDRESS_BEGIN.equals(code);
            return flag;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return false;
    }
}
